package com.flink.sink.mysql;

import com.flink.datasource.UserSource;
import com.flink.entity.CountUser;
import com.flink.entity.User;
import com.sequence.IdUtil;
import com.sequence.Sequence;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Calendar;

/**
 * 描述:
 * 输出到mysql
 *
 * @author yanzhengwu
 * @create 2022-07-23 22:51
 */
public class SinkToMysql {
    //com.flink.sink.mysql.SinkToMysql
    private static final String URL = "jdbc:mysql://192.168.2.209:3306/flink_data?characterEncoding=utf8";
    private static final String DRIVER_NAME = "com.mysql.jdbc.Driver";
    private static final String USERNAME = "root";
    private static final String password = "root";
    private static final Sequence SEQUENCE = new Sequence();

    public static void main(String[] args) throws Exception {
        //声明执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //测试为了保证元素顺序并行度设置为1，可以理解为单线程执行
        //env.setParallelism(1);


        SingleOutputStreamOperator<User> stream = env.addSource(new UserSource(500))
                //指定乱序流多等1秒
                .assignTimestampsAndWatermarks(WatermarkStrategy.<User>forBoundedOutOfOrderness(Duration.ofSeconds(1L))
                        .withTimestampAssigner(new SerializableTimestampAssigner<User>() {
                            @Override
                            public long extractTimestamp(User user, long recordTimestamp) {
                                return user.getTimestamp();
                            }
                        }));


        SingleOutputStreamOperator<CountUser> aggregate = stream.keyBy(User::getName)
                //10秒一个滑动窗口计算 多等1秒关闭窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(10), Time.seconds(1L)))
                .aggregate(new MyAggregateFunction(), new MyProcessWindowFunction());

        aggregate.print();
        aggregate.addSink(getSink());

        env.execute();

    }


    /**
     * TODO 聚合逻辑
     */
    public static class MyAggregateFunction implements AggregateFunction<User, Tuple2<Long, Long>, Tuple2<Long, Long>> {
        @Override
        public Tuple2<Long, Long> createAccumulator() {
            return Tuple2.of(0L, 0L);
        }

        /**
         * 分组后的其他维度统计逻辑可以在这里写，根据不通的状态分别统计数据，有几个字段就写几个元组类型即可
         * 这里所有的数据都已经是keBy分组后的数据
         *
         * @param value
         * @param accumulator
         * @return
         */
        @Override
        public Tuple2<Long, Long> add(User value, Tuple2<Long, Long> accumulator) {
            accumulator.f0 = accumulator.f0 + 1;
            if (value.getStatus() == 0) {
                accumulator.f1 = accumulator.f1 + 1;
            }
            return accumulator;
        }

        @Override
        public Tuple2<Long, Long> getResult(Tuple2<Long, Long> accumulator) {
            return accumulator;
        }

        @Override
        public Tuple2<Long, Long> merge(Tuple2<Long, Long> a, Tuple2<Long, Long> b) {
            return null;
        }
    }

    /**
     * TODO 处理逻辑
     */
    public static class MyProcessWindowFunction extends ProcessWindowFunction<Tuple2<Long, Long>, CountUser, String, TimeWindow> {
        @Override
        public void process(String key, Context context, Iterable<Tuple2<Long, Long>> elements, Collector<CountUser> out) throws Exception {
            CountUser countUser = new CountUser();
            Tuple2<Long, Long> val = elements.iterator().next();
            Long timestamp = Calendar.getInstance().getTimeInMillis();

            countUser.setTimeStamp(timestamp);
            countUser.setId(IdUtil.getId());
            countUser.setCounts(val.f0);
            countUser.setStatusCount(val.f1);
            countUser.setName(key);
            countUser.setWindowStart(context.window().getStart());
            countUser.setWindowEnd(context.window().getEnd());

            out.collect(countUser);


        }
    }

    /**
     * TODO 持久化逻辑
     *
     * @return
     */
    private static SinkFunction getSink() {
        return JdbcSink.<CountUser>sink(
                "insert into count_user (id,name,window_start, window_end,counts,status_counts,create_time,timestamp) values (?,?,?,?,?,?,?,?)",
                (statement, user) -> {
                    //System.out.println("拼接数据=====" + user.getTimestamp());
                    statement.setLong(1, SEQUENCE.nextId());
                    statement.setString(2, user.getName());
                    //System.out.println(user.getName());
                    statement.setLong(3, user.getWindowStart());
                    statement.setLong(4, user.getWindowEnd());
                    statement.setLong(5, user.getCounts());
                    statement.setLong(6, user.getStatusCount());
                    statement.setString(7, user.formatDate(user.getTimeStamp()));
                    statement.setLong(8, user.getTimeStamp());
                },
                JdbcExecutionOptions.builder()
                        //批量处理数量 TODO 默认为5000
                        .withBatchSize(1)
                        //批量处理间隔毫秒数  TODO 默认为0
                        //.withBatchIntervalMs(200)
                        //设置重试次数 TODO 默认为3
                        .withMaxRetries(5)
                        .build(),
                getJdbc()
        );

    }

    /**
     * TODO 读取jdbc连接
     *
     * @return
     */
    private static JdbcConnectionOptions getJdbc() {
        return new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl(URL)
                .withDriverName(DRIVER_NAME)
                .withUsername(USERNAME)
                .withPassword(password)
                .build();
    }
}
